'''
@Company: TWL
@Author: xue jian
@Email: xuejian@kanzhun.com
@Date: 2020-04-23 15:14:04
'''
# -*- coding: UTF-8 -*-
from kafka import KafkaProducer
from kafka import KafkaConsumer
from kafka import TopicPartition
import time, json, random
import threading

bootstrap_servers=['172.21.32.178:9092', '172.21.32.154:9092', '172.21.32.41:9092', '172.21.32.50:9092', '172.21.32.98:9092']
topic = "boss.arc.recommender.bossrec_sample_flow"
consumer = KafkaConsumer(topic, group_id='zrz', bootstrap_servers=bootstrap_servers)
consumer.unsubscribe()

partitions = consumer.partitions_for_topic(topic)
time_dict = {}
for part in partitions:
    time_dict[TopicPartition(topic, part)] = time.time()*1000  - 100 * 60 * 1000

ot_dict = consumer.offsets_for_times(time_dict)
print(ot_dict)
for k, v in ot_dict.items():
    if v != 'null':
        consumer.assign([k])
        consumer.seek(k, v[0])
        print(consumer.end_offsets([k]))
        last_offset = consumer.end_offsets([k])[k]
        no_consume_num = last_offset - v[0]
        print(no_consume_num)

i = 0
for msg in consumer:
#     print(msg.value)
    if i % 30 == 0:
        tp = TopicPartition(msg.topic, msg.partition)
        highwater = consumer.highwater(tp)
        # print(highwater)
        # print(msg.offset)
        lag = (highwater - 1) - msg.offset
        print('lag = ', lag)

    i += 1
#    time.sleep(1)


# consumer = KafkaConsumer(topic, group_id='zrz', bootstrap_servers=bootstrap_servers)
# consumer.unsubscribe()

# partitions = consumer.partitions_for_topic(topic)
# time_dict = {}
# for part in partitions:
#     time_dict[TopicPartition(topic, part)] = time.time()*1000  - 100 * 60 * 1000

# ot_dict = consumer.offsets_for_times(time_dict)
# for k, v in ot_dict.items():
#     if v != 'null':
#         consumer.assign([k])
#         consumer.seek(k, v[0])
#         last_offset = consumer.end_offsets([k])[k]
#         no_consume_num = last_offset - v[0]
#         print(no_consume_num)

# i = 0
# for msg in consumer:
#     if i > 3:
#         break
#     i += 1
#     print(msg.value)
#     time.sleep(1)